package org.midy.client;

import org.midy.infrastructure.tools.Logger;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

public class ClientMain {
    public static final Charset CHARSET_UTF8 = StandardCharsets.UTF_8;
    public static final int PRE_LENGTH = 6;

    public static void main(String[] args) {
        final LinkedBlockingQueue<String> msgQueue = new LinkedBlockingQueue<>(); // 待发送的消息队列
        final AtomicBoolean mainWait = new AtomicBoolean(true);
        final Selector selector = openSocketChannel("127.0.0.1", 8087, mainWait, msgQueue);
        final Thread thread = createConsoleInput(msgQueue, selector);

        holdMainThread(mainWait);
    }

    private static Thread createConsoleInput(LinkedBlockingQueue<String> msgQueue, Selector selector) {
        Thread thread = new Thread(() -> {
            try (BufferedReader reader = new BufferedReader(new InputStreamReader(System.in, StandardCharsets.UTF_8))) {
                while (true) {
                    String msg = reader.readLine(); // 从控制台读消息
                    msgQueue.put(msg);  // 待发送的消息入未发送消息队列
                    selector.wakeup(); // 通知监听器, 有待发送的消息
                    Logger.info("send: %s", msg);
                }
            } catch (Throwable t) {
                Logger.warn("读控制台信息失败：", t);
            }
        }, "CONSOLE-INPUT-THREAD");

        thread.setDaemon(true);
        thread.start();
        return thread;
    }

    public static Selector openSocketChannel(String hostName, int port, AtomicBoolean mainWaiting, LinkedBlockingQueue<String> msgQueue) {
        try {
            SocketChannel openSocketChannel = SocketChannel.open();
            openSocketChannel.socket().setKeepAlive(true);
            openSocketChannel.configureBlocking(false);
            Logger.info("waiting connect %s:%d", hostName, port);
            openSocketChannel.connect(new InetSocketAddress(hostName, port));
            return createChannelEventHandler(openSocketChannel, mainWaiting, msgQueue);
        } catch (IOException e) {
            Logger.error("创建SocketChannel失败：", e);
            throw new RuntimeException(e);
        } catch (Exception e) {
            Logger.error("初始化Channel线程任务失败：", e);
            throw new RuntimeException(e);
        }
    }

    public static Selector createChannelEventHandler(final SocketChannel openSocketChannel, final AtomicBoolean mainWaiting, LinkedBlockingQueue<String> msgQueue) throws IOException {
        final Selector selector = Selector.open();
        Thread thread = new Thread(() -> {
            try {
                openSocketChannel.register(selector, SelectionKey.OP_CONNECT);
                while (mainWaiting.get()) {

                    Logger.info("next selectedKeys ");
                    selector.select();

                    try {
                        Set<SelectionKey> selectionKeys = selector.selectedKeys();
                        Iterator<SelectionKey> iteratorSelectionKey = selectionKeys.iterator();
                        while (iteratorSelectionKey.hasNext()) {
                            SelectionKey currSelectionKey = iteratorSelectionKey.next();

                            if (!currSelectionKey.isValid()) { // 验证 时间监听Key 是否是有效的
                                Logger.warn("currSelectionKey.isValid()： false");
                                continue;
                            }

                            if (currSelectionKey.isValid() && currSelectionKey.isReadable()) { // 连接断开的情况下, 会触发一个读事件
                                Logger.info("新的Readable事件");
                                readProcess(currSelectionKey); // 读事件完成后, 由于是长连接应该继续监听读事件
                            }

                            if (currSelectionKey.isValid() && currSelectionKey.isWritable()) { // 注册了写的动作, 2.系统有写的条件了
                                Logger.info("新的Writable事件");
                                SocketChannel channel = (SocketChannel) currSelectionKey.channel();
                                currSelectionKey.interestOps(SelectionKey.OP_READ); // 写完处理完后, 应当注销写操作事件监听
                                writeProcess(channel, msgQueue);
                            }

                            if (currSelectionKey.isValid() && currSelectionKey.isConnectable()) {
                                Logger.info("新的Connectable事件");
                                SocketChannel channel = (SocketChannel) currSelectionKey.channel();
                                if (channel.finishConnect()) { // 连接失败会报错 TODO
                                    Logger.info("连接成功");
                                    channel.configureBlocking(false);
                                    channel.socket().setKeepAlive(true);
                                    currSelectionKey.interestOps(SelectionKey.OP_READ); // ‘连接事件’ -> '读事件'
                                }
                            }
                        }

                        if (!msgQueue.isEmpty()) { // 有待发送的新消息, 所以注册一个写事件
                            Iterator<SelectionKey> selectionKeyIterator = selector.keys().iterator();
                            while (selectionKeyIterator.hasNext()) {

                                SelectionKey currentSelectionKey = selectionKeyIterator.next();
                                Logger.info("hasNext： %d", currentSelectionKey.interestOps());

                                if ((currentSelectionKey.interestOps() & SelectionKey.OP_READ) == SelectionKey.OP_READ
                                        && (currentSelectionKey.interestOps() & ~SelectionKey.OP_WRITE) != SelectionKey.OP_WRITE) {
                                    Logger.info("向有效连接添加一个待写事件监听：%s", ((SocketChannel) currentSelectionKey.channel()).getRemoteAddress().toString());
                                    currentSelectionKey.interestOps(currentSelectionKey.interestOps() | SelectionKey.OP_WRITE);
                                }
                            }
                        }

                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                }
            } catch (ClosedChannelException e) {
                throw new RuntimeException(e);
            } catch (IOException e) {
                Logger.warn("Selector is close");
                throw new RuntimeException(e);
            }

        }, "CHANNEL-THREAD");
        thread.setDaemon(true);
        thread.start();
        return selector;
    }

    private static void readProcess(SelectionKey currSelectionKey) {
        SocketChannel channel = (SocketChannel) currSelectionKey.channel();
        if (!channel.isConnected()) {
            throw new RuntimeException("已断开连接");
        }
        try {
            ByteBuffer headBuffer = ByteBuffer.allocate(PRE_LENGTH);
            while (headBuffer.position() < headBuffer.capacity()) {
                int readSize = channel.read(headBuffer);
                if (readSize == -1 || readSize == 0) {
                    Logger.warn("读数据完毕");
                    break;
                }
                Logger.info("1 read %d byte...", readSize);
            }

            headBuffer.flip();

            if(headBuffer.limit()== 0) {
                return;
            }

            Integer bodySize = Integer.valueOf(new String(headBuffer.array(), CHARSET_UTF8));

            if (bodySize > 0) {
                ByteBuffer bodyBuffer = ByteBuffer.allocate(bodySize);
                while (bodyBuffer.position() < bodyBuffer.capacity()) {
                    int readSize = channel.read(bodyBuffer);
                    if (readSize == -1) {
                        Logger.warn("读数据完毕");
                        break;
                    }
                    Logger.info("2  read %d byte...", readSize);
                }
                bodyBuffer.flip();
                Logger.info("接受到信息：%s", new String(bodyBuffer.array(), CHARSET_UTF8));
            }
        } catch (Throwable t) {
            currSelectionKey.cancel();
            try {
                channel.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
            Logger.warn("读数据失败: %s", t.toString());
        }
    }

    private static void writeProcess(SocketChannel channel, LinkedBlockingQueue<String> msgQueue) {
        if (!channel.isConnected()) {
            throw new RuntimeException("已断开连接");
        }
        String msg = null;
        while ((msg = msgQueue.poll()) != null) {
            try { // 消息之间互不影响
                byte[] date = msg.getBytes(CHARSET_UTF8);
                ByteBuffer buffer = ByteBuffer.allocate(date.length + PRE_LENGTH);
                buffer.put(String.format("%06d", date.length).getBytes(CHARSET_UTF8));
                buffer.put(date);
                buffer.flip();
                channel.write(buffer);
            } catch (Throwable t) {
                try {
                    channel.close();
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
                Logger.warn("写信息失败：%s", msg);
            }
        }
    }

    public static void holdMainThread(final AtomicBoolean waiting) {
        Logger.info(" client started");
        while (waiting.get()) {
            Logger.info(" client waiting ...");
            try {
                synchronized (waiting) {
                    waiting.wait();
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
        Logger.info(" client closed");
    }
}
